package com.mall.socket;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.mall.redis.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ServerEndpoint 通过这个 spring boot 就可以知道你暴露出去的 ws 应用的路径，有点类似我们经常用的@RequestMapping。比如你的启动端口是8080，而这个注解的值是ws，那我们就可以通过 ws://127.0.0.1:8080/ws 来连接你的应用
 * @OnOpen 当 websocket 建立连接成功后会触发这个注解修饰的方法，注意它有一个 Session 参数
 * @OnClose 当 websocket 建立的连接断开后会触发这个注解修饰的方法，注意它有一个 Session 参数
 * @OnMessage 当客户端发送消息到服务端时，会触发这个注解修改的方法，它有一个 String 入参表明客户端传入的值
 * @OnError 当 websocket 建立连接时出现异常会触发这个注解修饰的方法，注意它有一个 Session 参数
 */
@ServerEndpoint("/webSocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
    @Autowired
    public RedisService redisService;
    public static RedisService staticRedisUtil;

    // 项目启动时把redisUtil交给静态的staticRedisUtil
    @PostConstruct
    public void init() {
        staticRedisUtil = this.redisService;
    }

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;
    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    //接收userId
    private String userId = "";

    //接收userId
    private static String KEY = "socketId";

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        //如果已经包含该用户id，则移除后重新加入
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            webSocketMap.put(userId, this);
        } else {
            //否则直接加入
            webSocketMap.put(userId, this);
        }
        addOnlineCount();           //在线数加1
        redisUnread(userId);
        log.info("有新窗口开始监听:" + userId + ",当前在线人数为" + getOnlineCount());
    }

    //redis 缓存消息
    public void redisUnread(String userId){
        //判断缓存中是否有用户未读消息
        String key = KEY+userId;
        boolean flag = staticRedisUtil.exists(key);
        if(flag){
            Long size = staticRedisUtil.size(key);
            List<Object> obj = staticRedisUtil.lRange(key,0,size);
            staticRedisUtil.remove(key);
            sendUnread(size,obj,userId);
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            log.info(">>> 用户：{}已关闭连接", userId);
        } else {
            log.info(">>> 连接已关闭...");
        }
        subOnlineCount();           //在线数减1
        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }


    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到来自窗口" + userId + "的信息:" + message);
        //群发消息
        Iterator<Map.Entry<String, WebSocketServer>> entries = webSocketMap.entrySet().iterator();
        while (entries.hasNext()) {
            Map.Entry<String, WebSocketServer> item = entries.next();
            item.getValue().sendMessage(message);
        }
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error(">>> WebSocket出现未知错误: ");
        error.printStackTrace();
    }

    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message){
        try {
            this.session.getBasicRemote().sendText(message);
        } catch (Exception e) {
            log.error(">>> WebSocket消息发送出现错误: ");
            e.printStackTrace();
        }
    }

    /**
     * 批量给用户发送消息
     */
    public static void sendMessageBatch(String message, List<String> userIdList) {
        userIdList.forEach(userId -> {
            sendInfo(message, userId);
        });
    }

    /**
     * 发送未读消息
     */
    public static void sendInfo(String message, @PathParam("userId") String userId){
        log.info("推送消息到窗口" + userId + "，推送内容:" + message);
        if(!StrUtil.isEmptyIfStr(userId)){
            if (webSocketMap.containsKey(userId)) {
                log.info(">>> 发送消息给：{}，内容：{}", userId, message);
                webSocketMap.get(userId).sendMessage(message);
            } else {
                log.error(">>> 用户：{}，不在线，不发送消息", userId);
                //redis缓存消息
                WebSocketServer.staticRedisUtil.lPush(KEY+userId,message);
            }
        }else{
            //群发消息
            Iterator<Map.Entry<String, WebSocketServer>> entries = webSocketMap.entrySet().iterator();
            while (entries.hasNext()) {
                Map.Entry<String, WebSocketServer> item = entries.next();
                item.getValue().sendMessage(message);
            }
            log.error(">>> 群发消息");
        }
    }

    //获取所有在线用户key
    public static List<String> getWebSocketMap(){
        List<String> list = new ArrayList<>();
        Iterator<Map.Entry<String, WebSocketServer>> entries = webSocketMap.entrySet().iterator();
        while (entries.hasNext()) {
            Map.Entry<String, WebSocketServer> item = entries.next();
            list.add(item.getKey());
        }
        return list;
    }

    //获取所有在线用户key
    public static int closeKey(String key){
        if (webSocketMap.containsKey(key)) {
            webSocketMap.remove(key);
            log.info(">>> 用户：{}已手动关闭连接", key);
        } else {
            log.info(">>> 连接已手动关闭...");
        }
        return 1;
    }

    /**
     * 群发自定义消息
     */
    public static void sendUnread(Long size,Object message, @PathParam("userId") String userId){
        log.info(">>> 发送未读消息给：{}，内容：{}", userId, message);
        Map<String,Object> map = new HashMap<>();
        map.put("size",size);
        map.put("msg",message);
        String msg = JSONUtil.toJsonStr(map);
        webSocketMap.get(userId).sendMessage(msg);
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}